package main

import (
	"bufio"
	"encoding/json"
	"fmt"
	"io/ioutil"
	"mapreduce-go/common"
	"mapreduce-go/master"
	"mapreduce-go/worker"
	"math/rand"
	"os"
	"regexp"
	"strconv"
	"strings"
	"time"
)

// 测试
// 词频统计
// 在若干文件中找到每个数字出现的次数
func main() {

	var config struct {
		JobName string   `json:"job_name"` // 任务名
		Master  string   `json:"master"`   // master ip
		Worker  []string `json:"worker"`   // worker ip
		InFile  []string `json:"in_file"`  // 输入文件
		NReduce int      `json:"n_reduce"` // reduce并行任务数量
		Random  struct {
			NNumber int `json:"n_number"`
			Count   int `json:"count"`
			RandTo  int `json:"rand_to"`
		} `json:"random"` // 生成随机测试数据
	}

	stat, err := os.Stat("./out")
	if err != nil || !stat.IsDir() {
		fmt.Println("out dir error : please check dir named out")
		return
	}

	fd, _ := ioutil.ReadFile("config.json")
	err = json.Unmarshal(fd, &config)
	if err != nil {
		fmt.Println("read config error : ", err.Error())
		return
	}

	// 创建输入文件
	if config.InFile == nil || len(config.InFile) == 0 {
		config.InFile = makeInputs(
			config.Random.NNumber,
			config.Random.Count,
			config.Random.RandTo,
		)
	}

	// 启动master
	mr := master.Distribute(config.JobName, config.InFile, config.NReduce, config.Master)

	// 启动worker
	for _, wkAddr := range config.Worker {
		go worker.RunWorker(config.Master, wkAddr, mapFunc, reduceFunc)
	}

	// 等待master完成信号
	<-mr.Done()
}

// 自定义的map分割函数
// line: 一行数据
func mapFunc(line string) []common.KeyValue {
	var res []common.KeyValue

	words := strings.Split(line, " ")
	for _, word := range words {
		if strings.TrimSpace(word) == "" {
			continue
		}

		// 只保留数字和字母
		word = strings.TrimSpace(word)
		reg := regexp.MustCompile(`[\W|_]{1,}`)
		word = reg.ReplaceAllString(word, "")

		kv := common.KeyValue{Key: word}
		res = append(res, kv)
	}

	if res == nil || len(res) == 0 {
		return []common.KeyValue{}
	}
	return res
}

// 自定义reduce函数
// key key
// value value集合
func reduceFunc(key string, value []string) string {
	return strconv.Itoa(len(value))
}

// 输入文件创建函数
// 根据指定的数量创建输入文件 并平均向每个文件写入数据 返回创建的文件名列表
// nNumber: 数据量
// count: 创建的文件数量
// randTo: 生成的随机数范围
func makeInputs(nNumber, count, randTo int) []string {
	var names []string
	var j = 0

	for i := 0; i < count; i++ {
		// 文件命名方式 根据mit6.824课程命名
		names = append(names, fmt.Sprintf("out/824-mrinput-%d.txt", i))

		file, err := os.Create(names[i])
		if err != nil {
			fmt.Println("error : ", err.Error())
		}

		writer := bufio.NewWriter(file)

		// 向每个mapFile中平均写入数据
		for j < (i+1)*(nNumber/count) {
			// 生成随机数
			rand.Seed(time.Now().UnixNano())
			random := rand.Intn(randTo)

			// 写入文件
			fmt.Fprintf(writer, "%d\n", random)

			j++
		}

		writer.Flush()
		file.Close()
	}

	return names
}
